package com.etl.component.basic;

import com.google.common.collect.Lists;
import com.component.api.constants.ComponentCodeConstants;
import com.component.api.constants.ComponentConstants;
import com.component.api.model.PortData;
import com.component.api.model.ProcessResult;
import com.component.api.model.data.ColumnData;
import com.component.api.model.data.SetData;
import com.component.api.model.dto.UnifieldColumnDTO;
import com.component.api.model.param.ParamPair;
import com.component.api.model.param.TableColumn;
import com.etl.component.common.AbstractFlinkComponent;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.Utils;
import org.apache.flink.api.java.operators.MapOperator;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.types.Row;

import java.util.List;
import java.util.Map;

/**
 * 描述：
 * 统一输出
 * @author xianggj
 * @Date 2021/12/3 11:20
 **/
public class UnificationOutputComponent extends AbstractFlinkComponent {

    @Override
    public ProcessResult process(List<PortData> datas, List<ParamPair> paramPairs) {
        Map<String, Object> paramMap = paramToObject(paramPairs);
        List<UnifieldColumnDTO> unifieldColumnDTOS =
                (List<UnifieldColumnDTO>) paramMap.get(ComponentConstants.COLUMNS);
        SetData input = (SetData)datas.get(0).getValue();
        DataSet<Row> dataset = (DataSet<Row>) input.getDataset();
        List<TableColumn> columns = input.getColumns();
        setIndex(unifieldColumnDTOS, columns);
        RowTypeInfo type = (RowTypeInfo)dataset.getType();
        TypeInformation<?>[] fieldTypes = type.getFieldTypes();
        DataSet<Row> result = new MapOperator<Row, Row>(dataset, type, e->{
            for (UnifieldColumnDTO unifieldColumnDTO : unifieldColumnDTOS) {
                int index = unifieldColumnDTO.getIndex();
                TypeInformation<?> fieldType =  fieldTypes[index];
                Object field = e.getField(index);
                if (field == null){
                    continue;
                }
                double val = Double.parseDouble(field.toString());
                Double resVal = unifieldColumnDTO.getFunction().calc(val);
                Object covert = null;
                //异常后默认为空
                try {
                    covert = covert(resVal, fieldType);
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
                e.setField(index, covert);
            }
            return e;
        }, Utils.getCallLocationName());
        return execute(result, columns,"执行统一输出操作");
    }

    @Override
    public List<PortData> previewOutputColumns(List<PortData> datas, List<ParamPair> paramPairs) {
        ColumnData input = (ColumnData)datas.get(0).getValue();
        List<TableColumn> columns = input.getColumns();
        return Lists.newArrayList(new PortData(getRule(), new ColumnData(columns)));
    }
    /**
     * 将某一列强行转换为int类型
     * @param i
     * @param dataset
     * @return
     */
    private RowTypeInfo changeType(int i, DataSet<Row> dataset) {
        RowTypeInfo type = (RowTypeInfo)dataset.getType();
        TypeInformation<?>[] fieldTypes = type.getFieldTypes();
        fieldTypes[i] = BasicTypeInfo.DOUBLE_TYPE_INFO;
        return new RowTypeInfo(fieldTypes, type.getFieldNames());
    }

    @Override
    public String getRule() {
        return ComponentCodeConstants.UNIFIED_OUTPUT;
    }
}
